package com.mpgame.database.redis.utils;

import com.google.protobuf.Message;
import com.google.protobuf.MessageLite;
import com.mpgame.utils.Log;
import com.mpgame.database.redis.config.RedisCfg;
import com.mpgame.database.redis.model.RedisCommand;
import com.mpgame.database.redis.model.RedisDBCtx;

import io.lettuce.core.RedisClient;
import io.lettuce.core.RedisURI;
import io.lettuce.core.TransactionResult;
import io.lettuce.core.api.StatefulRedisConnection;
import io.lettuce.core.api.sync.RedisCommands;
import io.lettuce.core.support.ConnectionPoolSupport;
import org.apache.commons.pool2.impl.GenericObjectPool;
import org.apache.commons.pool2.impl.GenericObjectPoolConfig;

import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

/**
 * Redis Lettuce客户端工具类
 * 
 * @author mpgame
 */
public class RedisUtil {

    private static GenericObjectPool<StatefulRedisConnection<String, String>> connectionPool;
    private static RedisCfg config;

    /**
     * 初始化Redis连接池
     * 
     * @param redisConfig Redis配置
     */
    public static void init(RedisCfg redisConfig) {
        config = redisConfig;

        // 构建Redis URI
        RedisURI.Builder uriBuilder = RedisURI.builder()
                .withHost(redisConfig.getHost())
                .withPort(redisConfig.getPort())
                .withDatabase(redisConfig.getDatabase())
                .withTimeout(Duration.ofMillis(redisConfig.getCommandTimeout()));

        if (redisConfig.getPassword() != null && !redisConfig.getPassword().isEmpty()) {
            uriBuilder.withPassword(redisConfig.getPassword());
        }

        RedisURI redisUri = uriBuilder.build();
        RedisClient redisClient = RedisClient.create(redisUri);

        // 配置连接池
        GenericObjectPoolConfig<StatefulRedisConnection<String, String>> poolConfig = new GenericObjectPoolConfig<>();
        poolConfig.setMaxTotal(redisConfig.getMaxTotal());
        poolConfig.setMaxIdle(redisConfig.getMaxIdle());
        poolConfig.setMinIdle(redisConfig.getMinIdle());
        poolConfig.setMaxWaitMillis(redisConfig.getMaxWaitMillis());
        poolConfig.setTestOnBorrow(redisConfig.isTestOnBorrow());
        poolConfig.setTestOnReturn(redisConfig.isTestOnReturn());
        poolConfig.setTestWhileIdle(redisConfig.isTestWhileIdle());

        // 创建连接池
        connectionPool = ConnectionPoolSupport.createGenericObjectPool(
                () -> redisClient.connect(), poolConfig);

        Log.info("Redis连接池初始化成功:" + redisConfig.getHost() + ":" + redisConfig.getPort());
    }

    private static final ConcurrentHashMap<Integer, RedisDBCtx> DB_CACHE = new ConcurrentHashMap<>();
    private static final int MAX_DB_CACHE_SIZE = 16;

    public static RedisDBCtx db(int dbIndex) {
        if (dbIndex < 0 || dbIndex >= MAX_DB_CACHE_SIZE) {
            throw new IllegalArgumentException("数据库索引必须在0-" + (MAX_DB_CACHE_SIZE - 1) + "之间");
        }
        return DB_CACHE.computeIfAbsent(dbIndex, RedisDBCtx::new);
    }

    /**
     * 获取Redis连接
     * 
     * @return Redis连接
     */
    public static StatefulRedisConnection<String, String> getConnection() {
        if (connectionPool == null) {
            throw new RuntimeException("Redis连接池未初始化，请先调用init方法");
        }
        try {
            return connectionPool.borrowObject();
        } catch (Exception e) {
            Log.err("获取Redis连接失败:" + e.getMessage());
            throw new RuntimeException("获取Redis连接失败", e);
        }
    }

    /**
     * 归还Redis连接
     * 
     * @param connection Redis连接
     */
    public static void returnConnection(StatefulRedisConnection<String, String> connection) {
        if (connection != null && connectionPool != null) {
            try {
                connectionPool.returnObject(connection);
            } catch (Exception e) {
                Log.err("归还Redis连接失败: " + e.getMessage());
            }
        }
    }

    /**
     * 执行Redis命令的通用方法
     * 
     * @param action 命令执行函数
     * @param <T>    返回类型
     * @return 执行结果
     */
    public static <T> T execute(RedisAction<T> action) {
        StatefulRedisConnection<String, String> connection = null;
        try {
            connection = getConnection();
            RedisCommands<String, String> commands = connection.sync();
            return action.execute(commands);
        } catch (Exception e) {
            Log.err("Redis命令执行失败:" + e.getMessage());
            throw new RuntimeException("Redis命令执行失败", e);
        } finally {
            returnConnection(connection);
        }
    }

    public static <T> T executeInDb(int dbIndex, RedisAction<T> action) {
        StatefulRedisConnection<String, String> connection = null;
        try {
            connection = getConnection();
            RedisCommands<String, String> commands = connection.sync();

            // 切换到指定数据库
            commands.select(dbIndex);

            // 执行操作
            return action.execute(commands);
        } catch (Exception e) {
            Log.err("Redis命令执行失败:" + e.getMessage());
            throw new RuntimeException("Redis命令执行失败", e);
        } finally {
            returnConnection(connection);
        }
    }

    /**
     * Redis命令执行接口
     * 
     * @param <T> 返回类型
     */
    @FunctionalInterface
    public interface RedisAction<T> {
        T execute(RedisCommands<String, String> commands);
    }

    // ==================== String 操作 ====================

    /**
     * 设置字符串值
     * 
     * @param key   键
     * @param value 值
     * @return 操作结果
     */
    public static String set(String key, String value) {
        return execute(commands -> commands.set(key, value));
    }

    /**
     * 设置字符串值（带过期时间）
     * 
     * @param key     键
     * @param value   值
     * @param seconds 过期时间（秒）
     * @return 操作结果
     */
    public static String setex(String key, String value, long seconds) {
        return execute(commands -> commands.setex(key, seconds, value));
    }

    /**
     * 获取字符串值
     * 
     * @param key 键
     * @return 值
     */
    public static String get(String key) {
        return execute(commands -> commands.get(key));
    }

    /**
     * 删除键
     * 
     * @param key 键
     * @return 删除的键数量
     */
    public static Long del(String key) {
        return execute(commands -> commands.del(key));
    }

    /**
     * 检查键是否存在
     * 
     * @param key 键
     * @return 是否存在
     */
    public static Boolean exists(String key) {
        Long res = execute(commands -> commands.exists(key));
        return res != null && res > 0;
    }

    /**
     * 设置键的过期时间
     * 
     * @param key     键
     * @param seconds 过期时间（秒）
     * @return 操作结果
     */
    public static Boolean expire(String key, long seconds) {
        return execute(commands -> commands.expire(key, seconds));
    }

    /**
     * 获取键的剩余过期时间
     * 
     * @param key 键
     * @return 剩余时间（秒），-1表示永不过期，-2表示键不存在
     */
    public static Long ttl(String key) {
        return execute(commands -> commands.ttl(key));
    }

    // ==================== Hash 操作 ====================

    /**
     * 设置Hash字段值
     * 
     * @param key   键
     * @param field 字段
     * @param value 值
     * @return 操作结果
     */
    public static Boolean hset(String key, String field, String value) {
        return execute(commands -> commands.hset(key, field, value));
    }

    /**
     * 获取Hash字段值
     * 
     * @param key   键
     * @param field 字段
     * @return 值
     */
    public static String hget(String key, String field) {
        return execute(commands -> commands.hget(key, field));
    }

    /**
     * 获取Hash所有字段和值
     * 
     * @param key 键
     * @return 字段值映射
     */
    public static Map<String, String> hgetall(String key) {
        return execute(commands -> commands.hgetall(key));
    }

    /**
     * 删除Hash字段
     * 
     * @param key    键
     * @param fields 字段
     * @return 删除的字段数量
     */
    public static Long hdel(String key, String... fields) {
        return execute(commands -> commands.hdel(key, fields));
    }

    /**
     * 检查Hash字段是否存在
     * 
     * @param key   键
     * @param field 字段
     * @return 是否存在
     */
    public static Boolean hexists(String key, String field) {
        return execute(commands -> commands.hexists(key, field));
    }

    /**
     * 设置Hash多个字段值
     * 
     * @param key         键
     * @param fieldValues 字段值映射
     * @return 操作结果
     */
    public static String hmset(String key, Map<String, String> fieldValues) {
        return execute(commands -> commands.hmset(key, fieldValues));
    }

    // ==================== List 操作 ====================

    /**
     * 从列表左侧推入元素
     * 
     * @param key    键
     * @param values 值
     * @return 列表长度
     */
    public static Long lpush(String key, String... values) {
        return execute(commands -> commands.lpush(key, values));
    }

    /**
     * 从列表右侧推入元素
     * 
     * @param key    键
     * @param values 值
     * @return 列表长度
     */
    public static Long rpush(String key, String... values) {
        return execute(commands -> commands.rpush(key, values));
    }

    /**
     * 从列表左侧弹出元素
     * 
     * @param key 键
     * @return 弹出的元素
     */
    public static String lpop(String key) {
        return execute(commands -> commands.lpop(key));
    }

    /**
     * 从列表右侧弹出元素
     * 
     * @param key 键
     * @return 弹出的元素
     */
    public static String rpop(String key) {
        return execute(commands -> commands.rpop(key));
    }

    /**
     * 获取列表指定范围的元素
     * 
     * @param key   键
     * @param start 开始位置
     * @param end   结束位置
     * @return 元素列表
     */
    public static List<String> lrange(String key, long start, long end) {
        return execute(commands -> commands.lrange(key, start, end));
    }

    /**
     * 获取列表长度
     * 
     * @param key 键
     * @return 列表长度
     */
    public static Long llen(String key) {
        return execute(commands -> commands.llen(key));
    }

    // ==================== Set 操作 ====================

    /**
     * 向集合添加成员
     * 
     * @param key     键
     * @param members 成员
     * @return 添加的成员数量
     */
    public static Long sadd(String key, String... members) {
        return execute(commands -> commands.sadd(key, members));
    }

    /**
     * 从集合移除成员
     * 
     * @param key     键
     * @param members 成员
     * @return 移除的成员数量
     */
    public static Long srem(String key, String... members) {
        return execute(commands -> commands.srem(key, members));
    }

    /**
     * 获取集合所有成员
     * 
     * @param key 键
     * @return 成员集合
     */
    public static Set<String> smembers(String key) {
        return execute(commands -> commands.smembers(key));
    }

    /**
     * 检查成员是否在集合中
     * 
     * @param key    键
     * @param member 成员
     * @return 是否存在
     */
    public static Boolean sismember(String key, String member) {
        return execute(commands -> commands.sismember(key, member));
    }

    /**
     * 获取集合成员数量
     * 
     * @param key 键
     * @return 成员数量
     */
    public static Long scard(String key) {
        return execute(commands -> commands.scard(key));
    }

    // ==================== Protobuf 序列化操作 ====================

    /**
     * 存储protobuf消息（推荐方式）
     * 
     * @param key     键
     * @param message protobuf消息
     * @return 操作结果
     */
    public static String setProtobuf(String key, MessageLite message) {
        try {
            byte[] data = message.toByteArray();
            String base64 = java.util.Base64.getEncoder().encodeToString(data);
            return set(key, base64);
        } catch (Exception e) {
            Log.err("Protobuf序列化失败:" + e.getMessage());
            return null;
        }
    }

    /**
     * 存储protobuf消息（带过期时间）
     * 
     * @param key     键
     * @param message protobuf消息
     * @param seconds 过期时间（秒）
     * @return 操作结果
     */
    public static String setexProtobuf(String key, MessageLite message, long seconds) {
        try {
            byte[] data = message.toByteArray();
            String base64 = java.util.Base64.getEncoder().encodeToString(data);
            return setex(key, base64, seconds);
        } catch (Exception e) {
            Log.err("Protobuf序列化失败: " + e.getMessage());
            return null;
        }
    }

    /**
     * 获取protobuf消息
     * 
     * @param key     键
     * @param builder protobuf消息构建器
     * @param <T>     消息类型
     * @return protobuf消息
     */
    public static <T extends MessageLite> T getProtobuf(String key, MessageLite.Builder builder) {
        String base64 = get(key);
        if (base64 == null) {
            return null;
        }
        try {
            byte[] data = java.util.Base64.getDecoder().decode(base64);
            return (T) builder.mergeFrom(data).build();
        } catch (Exception e) {
            Log.err("Protobuf反序列化失败: " + e.getMessage());
            return null;
        }
    }

    /**
     * 存储protobuf消息到Hash（适合需要部分更新的场景）
     * 
     * @param key     键
     * @param message protobuf消息
     */
    public static void setProtobufAsHash(String key, Message message) {
        try {
            // 将protobuf消息的所有字段存储为Hash
            Map<String, Object> fields = protobufToMap(message);
            for (Map.Entry<String, Object> entry : fields.entrySet()) {
                hset(key, entry.getKey(), String.valueOf(entry.getValue()));
            }
        } catch (Exception e) {
            Log.err("Protobuf Hash存储失败:" + e.getMessage());
        }
    }

    /**
     * 从Hash获取protobuf消息
     * 
     * @param key     键
     * @param builder protobuf消息构建器
     * @param <T>     消息类型
     * @return protobuf消息
     */
    public static <T extends Message> T getProtobufFromHash(String key, Message.Builder builder) {
        Map<String, String> hash = hgetall(key);
        if (hash.isEmpty()) {
            return null;
        }

        try {
            // 从Hash字段构建protobuf消息
            return (T) mapToProtobuf(hash, builder);
        } catch (Exception e) {
            Log.err("Protobuf Hash反序列化失败:" + e.getMessage());
            return null;
        }
    }

    /**
     * 将protobuf消息转换为Map（简化实现）
     */
    private static Map<String, Object> protobufToMap(Message message) {
        // 这里是一个简化的实现，实际项目中可能需要更复杂的反射处理
        Map<String, Object> map = new java.util.HashMap<>();
        map.put("data", java.util.Base64.getEncoder().encodeToString(message.toByteArray()));
        return map;
    }

    /**
     * 从Map构建protobuf消息（简化实现）
     */
    private static Message mapToProtobuf(Map<String, String> map, Message.Builder builder) throws Exception {
        String data = map.get("data");
        if (data != null) {
            byte[] bytes = java.util.Base64.getDecoder().decode(data);
            return builder.mergeFrom(bytes).build();
        }
        return builder.build();
    }

    // ==================== 原子操作 ====================

    /**
     * 原子递增
     * 
     * @param key 键
     * @return 递增后的值
     */
    public static Long incr(String key) {
        return execute(commands -> commands.incr(key));
    }

    /**
     * 原子递增指定值
     * 
     * @param key       键
     * @param increment 递增值
     * @return 递增后的值
     */
    public static Long incrBy(String key, long increment) {
        return execute(commands -> commands.incrby(key, increment));
    }

    /**
     * 原子递减
     * 
     * @param key 键
     * @return 递减后的值
     */
    public static Long decr(String key) {
        return execute(commands -> commands.decr(key));
    }

    /**
     * 原子递减指定值
     * 
     * @param key       键
     * @param decrement 递减值
     * @return 递减后的值
     */
    public static Long decrBy(String key, long decrement) {
        return execute(commands -> commands.decrby(key, decrement));
    }

    // ==================== 其他操作 ====================

    /**
     * 获取所有键
     * 
     * @param pattern 键模式
     * @return 键集合
     */
    public static List<String> keys(String pattern) {
        return execute(commands -> commands.keys(pattern));
    }

    /**
     * 清空当前数据库
     * 
     * @return 操作结果
     */
    public static String flushdb() {
        return execute(commands -> commands.flushdb());
    }

    /**
     * 清空所有数据库
     * 
     * @return 操作结果
     */
    public static String flushall() {
        return execute(commands -> commands.flushall());
    }

    /**
     * 获取数据库信息
     * 
     * @return 数据库信息
     */
    public static String info() {
        return execute(commands -> commands.info());
    }

    /**
     * 关闭连接池
     */
    public static void close() {
        if (connectionPool != null && !connectionPool.isClosed()) {
            connectionPool.close();
            Log.info("Redis连接池已关闭");
        }
    }

    public static TransactionResult pipeline(List<RedisCommand> commands) {
        return execute(redisCommands -> {

            // 开启管道
            redisCommands.multi();

            // 添加所有命令到管道
            for (RedisCommand cmd : commands) {
                switch (cmd.getType()) {
                    case HGETALL:
                        redisCommands.hgetall(cmd.getKey());
                        break;
                    case SISMEMBER:
                        redisCommands.sismember(cmd.getKey(), cmd.getValue());
                        break;
                    case GET:
                        redisCommands.get(cmd.getKey());
                        break;
                    case SET:
                        redisCommands.set(cmd.getKey(), cmd.getValue());
                        break;
                    case SADD:
                        redisCommands.sadd(cmd.getKey(), cmd.getValue());
                        break;
                    case SREM:
                        redisCommands.srem(cmd.getKey(), cmd.getValue());
                        break;
                    default:
                        Log.warn("未支持的Redis命令类型：" + cmd.getType());
                        break;
                }
            }

            // 执行管道并获取结果
            TransactionResult pipelineResults = redisCommands.exec();
            return pipelineResults;
        });
    }

    /**
     * 将Protobuf消息推入列表左侧
     */
    public static Long lpushProtobuf(String key, com.google.protobuf.MessageLite message) {
        return execute(commands -> {
            try {
                byte[] data = message.toByteArray();
                String base64 = java.util.Base64.getEncoder().encodeToString(data);
                return commands.lpush(key, base64);
            } catch (Exception e) {
                Log.err("Protobuf序列化失败: " + e.getMessage());
                return 0L;
            }
        });
    }

    /**
     * 从列表右侧弹出Protobuf消息
     */
    public static <T extends com.google.protobuf.MessageLite> T rpopProtobuf(String key, com.google.protobuf.MessageLite.Builder builder) {
        return execute(commands -> {
            try {
                String base64 = commands.rpop(key);
                if (base64 == null) {
                    return null;
                }
                byte[] data = java.util.Base64.getDecoder().decode(base64);
                return (T) builder.mergeFrom(data).build();
            } catch (Exception e) {
                Log.err("Protobuf反序列化失败: " + e.getMessage());
                return null;
            }
        });
    }
}